package pku;

import pku.DefaultMessage;
import pku.consumer.BufferService;
import pku.consumer.MessageReader;
import pku.serializers.MessageSerializer;


import java.io.File;
import java.util.ArrayList;
import java.util.Collection;

public class Consumer {

    private BufferService bufferService;
    private ArrayList<MessageReader> readers = new ArrayList<>();
    private int pollIndex = 0;
    private int count = 0;

    private MessageSerializer deserializer = new MessageSerializer(); // thread local
    
    public Consumer() {
        this.bufferService = BufferService.getInstance("data");
    }


    public DefaultMessage poll() {
        MessageReader reader = readers.get(pollIndex);
        DefaultMessage message = reader.readMessage();
        while (message == null) {
            readers.remove(pollIndex);
            if (readers.isEmpty()) return null;
            pollIndex = pollIndex % readers.size();
            reader = readers.get(pollIndex);
            message = reader.readMessage();
        }
        if ((++count & 0xff) == 0) { // change buffer every 64 messages
            pollIndex = (pollIndex + 1) % readers.size();

        }
        return message;
    }

    public void attachQueue(String queueName, Collection<String> topics) {
        ArrayList<String> fileNames = new ArrayList<>();
        for (String topic: topics) {
            for (int i = 0; i < 4; i++) {
                String name = topic + "_Thread-" + i;
                File f = new File("data" + File.separator + name);
                if (f.exists()) fileNames.add(name);
            }
        }
        for (String fileName: fileNames) {
            readers.add(new MessageReader(fileName, false, bufferService, deserializer));
        }
    }
}
